package com.dtruth.dataflow.bolt;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Map;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;

import com.dtruth.dataflow.util.MyLoggerFactory;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class DataValidateBolt extends BaseBasicBolt {

	private static Logger log = MyLoggerFactory
			.getLogger(DataValidateBolt.class);
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("news"));
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		@SuppressWarnings("unchecked")
		Map<String, Object> map = (Map<String, Object>) input.getValue(0);
		try {
			Object pubdate = map.get("pubdate");
			if (validatePubDate(pubdate)) {
				log.debug("emit:" + map.get("url"));
				collector.emit(new Values(map));
			} else {
				log.warn("pubdate is out of range :" + map.get("url"));
			}
		} catch (Exception e) {
			log.error(ExceptionUtils.getStackTrace(e) + ":" + map.get("url"));
		}

	}

	private boolean validatePubDate(Object pdObj) throws ParseException {
		if (pdObj == null)
			return false;
		long pubdate = Long.parseLong(pdObj.toString());
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
		Calendar c = Calendar.getInstance();
		c.add(Calendar.DATE, -1);
		long yesterday = sdf.parse(sdf.format(c.getTime())).getTime();
		long interval = pubdate - yesterday;
		if (interval >= 0)
			return true;
		return false;
	}
}